package com.zy.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.zy.gmall.realtime.app.func.BeanToJsonStrFunction;
import com.zy.gmall.realtime.bean.TradeSkuOrderBean;
import com.zy.gmall.realtime.common.GmallConfig;
import com.zy.gmall.realtime.util.*;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.async.RedisAsyncCommands;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.async.AsyncFunction;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.AsyncConnection;
import org.apache.hadoop.hbase.client.Connection;
import redis.clients.jedis.Jedis;

import java.math.BigDecimal;
import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

public class DwsTradeSkuOrderWindow {
    //从Kafka订单明细主题读取数据，过滤null数据并按照唯一键对数据去重，
    // 按照SKU维度分组，统计原始金额、活动减免金额、优惠券减免金额和订单金额，
    // 并关联维度信息，将数据写入 Doris 交易域SKU粒度下单各窗口汇总表
    public static void main(String[] args) {
        //创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        //设置检查点 开启 设置超时 最小间隔 重启策略 存活时长 状态后端 用户设置
        env.enableCheckpointing(50000L);

        //获取数据源
            //设置topic group
        String topic = "dwd_trade_order_detail";
        String groupId = "DwsTradeSkuOrderWindow";
        KafkaSource<String> kafkaSource = KafkaUtil.getKafkaSource(topic, groupId);
        DataStreamSource<String> kafkaSourceDs
                = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_source");
        //数据转换成JSONObj 并且过滤掉null数据
        SingleOutputStreamOperator<JSONObject> processFilterDs = kafkaSourceDs.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String str, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                JSONObject jsonObj = JSON.parseObject(str);
                if (!jsonObj.isEmpty()) {
                    collector.collect(jsonObj);
                }
            }
        });
        //processDs.print(">>>>");
        //{"create_time":"2023-08-31 18:37:39","sku_num":"1","split_original_amount":"8197.0000",
        // "split_coupon_amount":"0.0","sku_id":"10","date_id":"2023-08-31","user_id":"475",
        // "province_id":"16","sku_name":"Apple iPhone 12 (A2404) 64GB 蓝色 支持移动联通电信5G 双卡双待手机",
        // "id":"14303860","order_id":"69684","split_activity_amount":"0.0","split_total_amount":"8197.0","ts":"1693564659"}
        //分组
        KeyedStream<JSONObject, String> orderIdDs = processFilterDs.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject jsonObject) throws Exception {
                return jsonObject.getString("id");
            }
        });

        //去重 1 状态 + 定时器   2 状态 + 消除
        SingleOutputStreamOperator<JSONObject> processDs = orderIdDs.process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
            private ValueState<JSONObject> valeState;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<JSONObject> StateDescriptor = new ValueStateDescriptor<>("StateDescriptor", JSONObject.class);
                StateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.seconds(10)).build());
                valeState = getRuntimeContext().getState(StateDescriptor);
            }

            @Override
            public void processElement(JSONObject jsonObj, KeyedProcessFunction<String, JSONObject, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                //消除 是判断状态中是否存在，不存在就是自己放进去，存在的话把必要的属性改为负值
                JSONObject lastValue = valeState.value();
                if (lastValue != null) {
                    //不为空，将取值边为负数
                    String splitOriginalAmount = lastValue.getString("split_original_amount");
                    String splitCouponAmount = lastValue.getString("split_coupon_amount");
                    String splitActivityAmount = lastValue.getString("split_activity_amount");
                    String splitTotalAmount = lastValue.getString("split_total_amount");
                    lastValue.put("split_original_amount", "-" + splitOriginalAmount);
                    lastValue.put("split_coupon_amount", "-" + splitCouponAmount);
                    lastValue.put("split_activity_amount", "-" + splitActivityAmount);
                    lastValue.put("split_total_amount", "-" + splitTotalAmount);
                    collector.collect(lastValue);
                }
                //传向下游
                collector.collect(jsonObj);
                //如果为空
                valeState.update(jsonObj);
            }
        });
        //processDs.print();

        //转换为实体类
        SingleOutputStreamOperator<TradeSkuOrderBean> mapDs = processDs.map(new MapFunction<JSONObject, TradeSkuOrderBean>() {
            @Override
            public TradeSkuOrderBean map(JSONObject jsonObject) throws Exception {
                return TradeSkuOrderBean.builder()
                        .ts(jsonObject.getLong("ts") * 1000)
                        .skuId(jsonObject.getString("sku_id"))
                        .originalAmount(new BigDecimal(jsonObject.getString("split_original_amount")))
                        .couponAmount(new BigDecimal(jsonObject.getString("split_coupon_amount")))
                        .activityAmount(new BigDecimal(jsonObject.getString("split_activity_amount")))
                        .orderAmount(new BigDecimal(jsonObject.getString("split_total_amount")))
                        .build();
            }
        });
        //获取水位线
        SingleOutputStreamOperator<TradeSkuOrderBean> waterMarkDs = mapDs.assignTimestampsAndWatermarks(
                WatermarkStrategy.<TradeSkuOrderBean>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<TradeSkuOrderBean>() {
                                    @Override
                                    public long extractTimestamp(TradeSkuOrderBean tradeSkuOrderBean, long l) {
                                        return tradeSkuOrderBean.getTs();
                                    }
                                }
                        )
        );
        //分组按照sku
        KeyedStream<TradeSkuOrderBean, String> keyByDs = waterMarkDs.keyBy(new KeySelector<TradeSkuOrderBean, String>() {
            @Override
            public String getKey(TradeSkuOrderBean tradeSkuOrderBean) throws Exception {
                return tradeSkuOrderBean.getSkuId();
            }
        });
        //开窗
        WindowedStream<TradeSkuOrderBean, String, TimeWindow> windowDs = keyByDs.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(5)));
        //聚合
        SingleOutputStreamOperator<TradeSkuOrderBean> reduceDs = windowDs.reduce(
                new ReduceFunction<TradeSkuOrderBean>() {

                    @Override
                    public TradeSkuOrderBean reduce(TradeSkuOrderBean t1, TradeSkuOrderBean t2) throws Exception {
                        t1.setActivityAmount(t1.getActivityAmount().add(t2.getActivityAmount()));
                        t1.setOrderAmount(t1.getOrderAmount().add(t2.getOrderAmount()));
                        t1.setCouponAmount(t1.getCouponAmount().add(t2.getCouponAmount()));
                        t1.setOriginalAmount(t1.getOriginalAmount().add(t2.getOriginalAmount()));
                        return t1;
                    }
                }
                ,
                new WindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>() {
                    @Override
                    public void apply(String str, TimeWindow timeWindow, Iterable<TradeSkuOrderBean> iterable, Collector<TradeSkuOrderBean> collector) throws Exception {
                        long start = timeWindow.getStart();
                        String startDate = DateFormatUtil.toYmdHms(start);
                        String curDate = DateFormatUtil.toDate(start);
                        long end = timeWindow.getEnd();
                        String endDate = DateFormatUtil.toYmdHms(end);

                        for (TradeSkuOrderBean orderBean : iterable) {
                            orderBean.setStt(startDate);
                            orderBean.setEdt(endDate);
                            orderBean.setCurDate(curDate);
                            collector.collect(orderBean);
                        }
                    }
                }
        );

        //reduceDs.print(">>>>");

        //进行和维度表关联
        //1 直接从Hbase直接去获取
//        reduceDs.map(
//                new RichMapFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
//                    Connection conn;
//                    @Override
//                    public void open(Configuration parameters) throws Exception {
//                        conn = HbaseUtil.getConnect();
//                    }
//
//                    @Override
//                    public void close() throws Exception {
//                        HbaseUtil.closeHbase(conn);
//                    }
//
//                    @Override
//            public TradeSkuOrderBean map(TradeSkuOrderBean tradeSkuOrderBean) throws Exception {
//                //从Hbase中获取到数据，然后把数据补充到实体类中
//                String skuId = tradeSkuOrderBean.getSkuId();
//                JSONObject dimSkuInfo = HbaseUtil.getDimInfoFromHbase(conn, GmallConfig.HBASE_NAMESPACE, "dim_sku_info", skuId);
//                    tradeSkuOrderBean.setSkuName(dimSkuInfo.getString("sku_name"));
//                    tradeSkuOrderBean.setSpuId(dimSkuInfo.getString("spu_id"));
//                    tradeSkuOrderBean.setTrademarkId(dimSkuInfo.getString("tm_id"));
//                    tradeSkuOrderBean.setCategory3Id(dimSkuInfo.getString("category3_id"));
//                return tradeSkuOrderBean;
//            }
//        }).print();

        //2 旁路缓存优化

//        reduceDs.map(
//                new RichMapFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
//                    private Connection conn;
//                    private Jedis jedis;
//                    @Override
//                    public void open(Configuration parameters) throws Exception {
//                        conn = HbaseUtil.getConnect();
//                        jedis = RedisUtil.getJedis();
//                    }
//
//                    @Override
//                    public void close() throws Exception {
//                        HbaseUtil.closeHbase(conn);
//                        RedisUtil.closeJedis(jedis);
//                    }
//
//                    @Override
//                    public TradeSkuOrderBean map(TradeSkuOrderBean tradeSkuOrderBean) throws Exception {
//                        //从Hbase中获取到数据，然后把数据补充到实体类中
//                        String skuId = tradeSkuOrderBean.getSkuId();
//                        JSONObject dimSkuInfo = DimUtil.getDimInfo(jedis,conn, GmallConfig.HBASE_NAMESPACE, "dim_sku_info", skuId);
//                        tradeSkuOrderBean.setSkuName(dimSkuInfo.getString("sku_name"));//！！！！！！！！！！！！注意排除null
//                        tradeSkuOrderBean.setSpuId(dimSkuInfo.getString("spu_id"));
//                        tradeSkuOrderBean.setTrademarkId(dimSkuInfo.getString("tm_id"));
//                        tradeSkuOrderBean.setCategory3Id(dimSkuInfo.getString("category3_id"));
//                        return tradeSkuOrderBean;
//                    }
//                }).print();

        //3 使用模板类试一试

        //4 异步IO方式
        SingleOutputStreamOperator<TradeSkuOrderBean> resultStream = AsyncDataStream.unorderedWait(reduceDs,
                new RichAsyncFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
                    StatefulRedisConnection<String, String> redisConn;
                    AsyncConnection hbaseConn;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //获取redis Hbase一异步连接
                        redisConn = RedisUtil.getAsyncRedisConnection();
                        hbaseConn = HbaseUtil.getAsyncConnection();
                    }

                    @Override
                    public void close() throws Exception {
                        //关闭redis Hbase异步连接
                        RedisUtil.closeAsyncRedisConn(redisConn);
                        HbaseUtil.closeAsyncConnection(hbaseConn);
                    }

                    @Override
                    public void asyncInvoke(TradeSkuOrderBean tradeSkuOrderBean, ResultFuture<TradeSkuOrderBean> resultFuture) throws Exception {
                        //分三步，这是第一步,在redis中查找
                        CompletableFuture.supplyAsync(
                                        new Supplier<JSONObject>() {
                                            @Override
                                            public JSONObject get() {
                                                String skuId = tradeSkuOrderBean.getSkuId();
                                                JSONObject jsonObj = RedisUtil.asyncGetDimInfo(redisConn, "dim_sku_info:" + skuId);
                                                System.out.println("redis不知道为不为空");
                                                return jsonObj;
                                            }
                                        }
                                )//第二步 ，判断是否
                                .thenApplyAsync(new Function<JSONObject, JSONObject>() {
                                    @Override
                                    public JSONObject apply(JSONObject jsonObj) {
                                        if (jsonObj == null) {
                                            System.out.println("果然为空,用hbase");
                                            jsonObj = HbaseUtil.getDimInfoFromHbaseByAsync(hbaseConn, GmallConfig.HBASE_NAMESPACE, "dim_sku_info", tradeSkuOrderBean.getSkuId());
                                        }
                                        return jsonObj;

                                    }
                                }).thenAcceptAsync(
                                        new Consumer<JSONObject>() {
                                            @Override
                                            public void accept(JSONObject jsonObj) {
                                                //join一下
                                                if (jsonObj != null) {
                                                    //将维度对象的属性补充到流中对象
                                                    System.out.println("补充数据");
                                                    tradeSkuOrderBean.setSkuName(jsonObj.getString("sku_name"));
                                                    tradeSkuOrderBean.setSpuId(jsonObj.getString("spu_id"));
                                                    tradeSkuOrderBean.setTrademarkId(jsonObj.getString("tm_id"));
                                                    tradeSkuOrderBean.setCategory3Id(jsonObj.getString("category3_id"));
                                                    //获取数据库交互的结果并发送给 ResultFuture的回调函数
                                                    resultFuture.complete(Collections.singleton(tradeSkuOrderBean));
                                                }
                                            }
                                        }
                                );
                    }
                },
                60,
                TimeUnit.SECONDS);

        resultStream.print(">>>");
        //还要关联几个维度表，把剩余的空值补充完全



//        resultStream
//                .map(new BeanToJsonStrFunction<>())
//                .sinkTo(DorisUtil.getDorisSink("dws_trade_sku_order_window"));


        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }
}